package xy.rocketmq.service;

import com.alibaba.fastjson.JSON;
import com.sun.xml.internal.ws.assembler.MetroTubelineAssembler;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * @Classname MQProducerService
 * @Description TODO
 * @Date 2023/7/14 15:30
 * @Created by wangxianlin
 */
@Slf4j
@Component
public class MQProducerService {

    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer messageTimeOut;

    // 建议正常规模项目统一用一个TOPIC
    private static final String topic = "TEST_TOPIC";

    // 直接注入使用，用于发送消息到broker服务器
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送普通的消息
     * @param msg
     */
    public void send(String msg){
        rocketMQTemplate.convertAndSend(topic+":tag1",msg);
    }

    /**
     * 发送同步消息（会阻塞当前线程，等待broker响应发送结果，这样不太容易丢失数据）
     * @param msg
     * @return
     */
    public SendResult sendSyncMsg(String msg){
        SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msg).build());
        log.info("【sendMsg】sendResult={}", JSON.toJSONString(sendResult));
        return sendResult;
    }

    /**
     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调，在sendCallback中可处理相关成功和失败的逻辑）
     * 适合 对响应时间敏感的业务场景
     * @param msg
     */
    public void sendAsyncMsg(String msg){
        rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(msg).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步发送消息成功");
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("异步发送消息异常啦");
            }
        });
    }

    /**
     * 发送延时消息（上面发送的是同步消息，delayLevel 的值为0，因为不延时）
     * 在start 版本中，延时消息一共分为： 1s ,5s ,10s ,30s 1m 2m 3m 4m 5m - 10m 20m 30m 1h 2h
     * @param msg
     * @param delayLevel
     */
    public void sendDelayMsg(String msg,int delayLevel){
        rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msg).build(),messageTimeOut,delayLevel);
    }

    /**
     * 发送单向消息（只负责发送消息，不等待应答，不关心发送结果）
     * @param msg
     */
    public void sendOneMsg(String msg){
        rocketMQTemplate.sendOneWay(topic,MessageBuilder.withPayload(msg).build());
    }

    /**
     * 发送带 tag的消息 直接在topic后面加上 :tag即可
     * @param msg
     * @return
     */
    public SendResult sendTagMsg(String msg){
        return rocketMQTemplate.syncSend(topic+":tag2",MessageBuilder.withPayload(msg).build());
    }
}
